package com.cl.spark.node;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import com.cl.spark.util.OKHttpUtil;
import org.apache.spark.sql.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Component
public class HttpRequestNode extends BaseSparkNode {
    @Autowired
    SparkSession sparkSession;

    @Override
    public SparkResult process(SparkParam sparkParam) {
        JSONObject expressionObj = sparkParam.getNodeExpression();
        String method = expressionObj.getString("method");
        String url = expressionObj.getString("url");
        String paramTemplate = expressionObj.getString("paramTemplate");
        String resultPosition = expressionObj.getString("resultPosition");
        Dataset<Row> dataset = sparkParam.getSparkResultList().get(0).getDataset();
        if (dataset.isEmpty()) {
            return SparkResult.success(dataset);
        }
        List<Row> newRowList = new ArrayList<>();
        for (Row row : dataset.collectAsList()) {
            List<JSONObject> resList = new ArrayList<>();
            try {
                Object resItem = new Object();
                String paramString = "";
                if (paramTemplate != null) {
                    paramString = paramTemplate;
                    Pattern pattern = Pattern.compile("\\$\\{(.+?)}");
                    Matcher matcher = pattern.matcher(paramTemplate);
                    while (matcher.find()) {
                        String fieldName = matcher.group(1);
                        paramString = paramString.replace(matcher.group(0), row.getAs(fieldName));
                    }
                }

                String resString = "";
                if ("POST".equals(method)) {
                    resString = OKHttpUtil.okhttpPostString(url, JSONObject.parseObject(paramString));
                } else if ("GET".equals(method)) {
                    resString = OKHttpUtil.okhttpGetString(url, JSONObject.parseObject(paramString));
                }
                if (JSONObject.isValid(resString)) {
                    JSONObject resObj = JSONObject.parseObject(resString);
                    if (StringUtils.hasLength(resultPosition) && resultPosition.contains(".")) {
                        String[] split = resultPosition.split("\\.");
                        JSONObject resItemObj = JSONObject.parseObject(resString);
                        for (int j = 0; j < split.length; j++) {
                            if (j == split.length - 1) {
                                resItem = resItemObj.get(split[j]);
                            } else {
                                resItemObj = resItemObj.getJSONObject(split[j]);
                            }
                        }
                    } else if (StringUtils.hasLength(resultPosition)) {
                        resItem = resObj.get(resultPosition);
                    } else {
                        resItem = resObj;
                    }
                    String resItemJSON = JSON.toJSONString(resItem);
                    if (JSONArray.isValidArray(resItemJSON)) {
                        JSONArray jsonArray = JSON.parseArray(resItemJSON);
                        if (!jsonArray.isEmpty()) {
                            for (int i = 0; i < jsonArray.size(); i++) {
                                resList.add(jsonArray.getJSONObject(i));
                            }
                        }
                    } else {
                        resList.add(JSON.parseObject(resItemJSON));
                    }
                }

                for (JSONObject jsonObject : resList) {
                    List<Object> values = new ArrayList<>();
                    for (int i = 0; i < row.length(); i++) {
                        values.add(row.get(i));
                    }
                    if (jsonObject != null) {
                        jsonObject.forEach((key, value) -> {
                            values.add(String.valueOf(value));
                        });
                    }
                    Row newRow = RowFactory.create(values.toArray(new Object[0]));
                    newRowList.add(newRow);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
        if (newRowList!= null && !newRowList.isEmpty()) {
            try {
                JSONObject first = JSONObject.parseObject(newRowList.get(0).json());
                for (Map.Entry<String, Object> entry : first.entrySet()) {
                    String key = entry.getKey();
                    dataset = dataset.withColumn(key, functions.lit(""));
                }
            }catch (Exception e) {
                e.printStackTrace();
            }

        }

        Dataset<Row> newDataset = sparkSession.createDataset(newRowList, dataset.encoder());

        return SparkResult.success(newDataset);
    }

    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.HTTP_REQUEST;
    }
}
